package com.kool.kmetrics.metrics;

import com.alibaba.fastjson.JSON;
import com.kool.kmetrics.client.RedisClient;
import com.kool.kmetrics.common.constant.ErrorCode;
import com.kool.kmetrics.common.exception.AppException;
import com.kool.kmetrics.common.util.DateUtil;
import com.kool.kmetrics.common.util.SpringUtil;
import com.kool.kmetrics.common.util.StringUtil;
import com.kool.kmetrics.consumer.vo.ReceiveLog;

import java.math.BigDecimal;
import java.text.ParseException;

import static com.kool.kmetrics.constant.MetricsRedisKey.*;

/**
 * 统计当日各种报文类型的入站报文数、成功数、失败数、平均耗时、最大耗时；
 * 当日用户的各种报文类型的入站报文数、成功数、失败数、平均耗时、最大耗时；
 * 当日客户端的各种报文类型的入站报文数、成功数、失败数、平均耗时、最大耗时；
 * 当日客户端地址的各种报文类型的入站报文数、成功数、失败数、平均耗时、最大耗时；
 * 当日主题的入站PUBLISH报文数；
 * 当日用户的主题的入站PUBLISH报文数；
 * 当日客户端的主题的入站PUBLISH报文数；
 * 当日客户端地址的主题的入站PUBLISH报文数;
 *
 * @author luyu
 */
public class ReceiveComputer implements MetricsComputer {
    private RedisClient redisClient = SpringUtil.getBean(RedisClient.class);
    /**
     * key过期时长，单位毫秒
     * 3天
     */
    public static final long TIMEOUT_DURING = 259200000L;

    /**
     * hash结构
     * 参数1：日期
     * hkey：topicName
     * hvalue:空
     */
    public static final String KEY_TODAY_TOPIC = "kmetrics:today_topic:%s";
    /**
     * hash结构
     * 参数1：日期
     * hkey：userName
     * hvalue:空
     */
    public static final String KEY_TODAY_USER = "kmetrics:today_user:%s";
    /**
     * hash结构
     * 参数1：日期
     * hkey：clientId
     * hvalue:空
     */
    public static final String KEY_TODAY_CLIENT_ID = "kmetrics:today_client_id:%s";
    /**
     * hash结构
     * 参数1：日期
     * hkey：remoteAddress
     * hvalue:空
     */
    public static final String KEY_TODAY_REMOTE_ADDRESS = "kmetrics:today_remote_address:%s";

    @Override
    public void compute(String log) {
        ReceiveLog receiveLog = JSON.parseObject(log, ReceiveLog.class);
        String time = receiveLog.getTimestamp();
        String packetType = receiveLog.getPacketType();
        String isSuccess = receiveLog.getIsSuccess();
        Long costTime = receiveLog.getCostTime();
        String userName = receiveLog.getUserName();
        String clientId = receiveLog.getClientId();
        String remoteAddress = receiveLog.getRemoteAddress();
        String topicName = receiveLog.getTopicName();

        String now = null;
        try {
            now = Integer.toString(DateUtil.toIntDay(DateUtil.parseDate(time, DateUtil.DateFormatEnum.DEFAULT.getFormat())));
        } catch (ParseException e) {
            throw new AppException(ErrorCode.BUSINESS_ERROR, e, e.getMessage());
        }

        //成功数
        if (isSuccess.equals("1")) {
            //平台维度
            redisClient.inrc(String.format(KEY_TODAY_RECEIVE_SUCCESS, now, packetType, "global", "global"), TIMEOUT_DURING);
            //主题PUBLISH数
            redisClient.inrc(String.format(KEY_TODAY_TOPIC_RECEIVE_SUCCESS, now, topicName, "global", "global"), TIMEOUT_DURING);
            //用户维度
            if (StringUtil.isNotEmpty(userName)) {
                redisClient.inrc(String.format(KEY_TODAY_RECEIVE_SUCCESS, now, packetType, "user_name", userName), TIMEOUT_DURING);
                redisClient.inrc(String.format(KEY_TODAY_TOPIC_RECEIVE_SUCCESS, now, topicName, "user_name", userName), TIMEOUT_DURING);
            }
            //客户端ID维度
            if (StringUtil.isNotEmpty(clientId)) {
                redisClient.inrc(String.format(KEY_TODAY_RECEIVE_SUCCESS, now, packetType, "client_id", clientId), TIMEOUT_DURING);
                redisClient.inrc(String.format(KEY_TODAY_TOPIC_RECEIVE_SUCCESS, now, topicName, "client_id", clientId), TIMEOUT_DURING);
            }
            //客户端地址维度
            if (StringUtil.isNotEmpty(remoteAddress)) {
                redisClient.inrc(String.format(KEY_TODAY_RECEIVE_SUCCESS, now, packetType, "remoteAddress", remoteAddress), TIMEOUT_DURING);
                redisClient.inrc(String.format(KEY_TODAY_TOPIC_RECEIVE_SUCCESS, now, topicName, "remoteAddress", remoteAddress), TIMEOUT_DURING);
            }

            //今日topic
            redisClient.hput(String.format(KEY_TODAY_TOPIC, now), topicName, "");
        }
        //失败数
        if (isSuccess.equals("0")) {
            //平台维度
            redisClient.inrc(String.format(KEY_TODAY_RECEIVE_FAIL, now, packetType, "global", "global"), TIMEOUT_DURING);
            //用户维度
            if (StringUtil.isNotEmpty(userName)) {
                redisClient.inrc(String.format(KEY_TODAY_RECEIVE_FAIL, now, packetType, "user_name", userName), TIMEOUT_DURING);
            }
            //客户端ID维度
            if (StringUtil.isNotEmpty(clientId)) {
                redisClient.inrc(String.format(KEY_TODAY_RECEIVE_FAIL, now, packetType, "client_id", clientId), TIMEOUT_DURING);
            }
            //客户端地址维度
            if (StringUtil.isNotEmpty(remoteAddress)) {
                redisClient.inrc(String.format(KEY_TODAY_RECEIVE_FAIL, now, packetType, "remoteAddress", remoteAddress), TIMEOUT_DURING);
            }
        }
        //总数
        //平台维度
        redisClient.inrc(String.format(KEY_TODAY_RECEIVE_TOTAL, now, packetType, "global", "global"), TIMEOUT_DURING);
        //用户维度
        if (StringUtil.isNotEmpty(userName)) {
            redisClient.inrc(String.format(KEY_TODAY_RECEIVE_TOTAL, now, packetType, "user_name", userName), TIMEOUT_DURING);
            //今日用户
            redisClient.hput(String.format(KEY_TODAY_USER, now), userName, "");
        }
        //客户端ID维度
        if (StringUtil.isNotEmpty(clientId)) {
            redisClient.inrc(String.format(KEY_TODAY_RECEIVE_TOTAL, now, packetType, "client_id", clientId), TIMEOUT_DURING);
            //今日客户
            redisClient.hput(String.format(KEY_TODAY_CLIENT_ID, now), clientId, "");
        }
        //客户端地址维度
        if (StringUtil.isNotEmpty(remoteAddress)) {
            redisClient.inrc(String.format(KEY_TODAY_RECEIVE_TOTAL, now, packetType, "remoteAddress", remoteAddress), TIMEOUT_DURING);
            //今日客户端地址
            redisClient.hput(String.format(KEY_TODAY_REMOTE_ADDRESS, now), remoteAddress, "");
        }


        //耗时均值和最大值
        if (isSuccess.equals("1")
                && costTime != null) {
            //均值
            //平台维度
            computeCostTimeAvg(now, packetType, "global", "global", costTime);
            //最大值
            computeCostTimeMax(now, packetType, "global", "global", costTime);
            //用户维度
            if (StringUtil.isNotEmpty(userName)) {
                computeCostTimeAvg(now, packetType, "user_name", userName, costTime);
                computeCostTimeMax(now, packetType, "user_name", userName, costTime);
            }
            //客户端ID维度
            if (StringUtil.isNotEmpty(clientId)) {
                computeCostTimeAvg(now, packetType, "client_id", clientId, costTime);
                computeCostTimeMax(now, packetType, "client_id", clientId, costTime);
            }
            //客户端地址维度
            if (StringUtil.isNotEmpty(remoteAddress)) {
                computeCostTimeAvg(now, packetType, "remoteAddress", remoteAddress, costTime);
                computeCostTimeMax(now, packetType, "remoteAddress", remoteAddress, costTime);
            }
        }
    }

    /**
     * 求当日耗时最大值
     *
     * @param now
     * @param packetType
     * @param entity
     * @param costTime
     */
    private void computeCostTimeMax(String now, String packetType, String entityType, String entity, Long costTime) {
        String max = redisClient.get(String.format(KEY_TODAY_RECEIVE_COST_MAX, now, packetType, entityType, entity));
        if (max == null) {
            redisClient.set(String.format(KEY_TODAY_RECEIVE_COST_MAX, now, packetType, entityType, entity), costTime.toString(), TIMEOUT_DURING);

        } else {
            long maxNum = Long.parseLong(max);
            if (maxNum < costTime) {
                //最好做一下CAS，不然会出现丢失最大值的情况，有必要这么准吗？ todo
                redisClient.set(String.format(KEY_TODAY_RECEIVE_COST_MAX, now, packetType, entityType, entity), costTime.toString(), TIMEOUT_DURING);
            }
        }
    }

    /**
     * 求当日耗时均值，sum和成功数没有做数据一致性，存在一定误差，允许了吧
     *
     * @param now
     * @param packetType
     * @param entity
     * @param costTime
     */
    private void computeCostTimeAvg(String now, String packetType, String entityType, String entity, Long costTime) {
        long sum = redisClient.inrc(String.format(KEY_TODAY_RECEIVE_COST_SUM, now, packetType, entityType, entity), costTime, TIMEOUT_DURING);
        String success = redisClient.get(String.format(KEY_TODAY_RECEIVE_SUCCESS, now, packetType, entityType, entity));
        long successNum = success == null ? 0L : Long.valueOf(success);
        if (successNum > 0) {
            //计算平均耗时,单位毫秒
            BigDecimal avg = new BigDecimal(sum).divide(new BigDecimal(successNum), 0, BigDecimal.ROUND_HALF_UP);
            redisClient.set(String.format(KEY_TODAY_RECEIVE_COST_AVG, now, packetType, entityType, entity), avg.toString(), TIMEOUT_DURING);
        }
    }
}
